package com.ceair.kafka.consumer;

import com.ceair.entity.LogExt;
import com.ceair.entity.SysLog;
import com.ceair.entity.dto.SysLogDto;
import com.ceair.mongodb.LogExtRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ceair.service.ISysLogService;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

/**
 * @author wangbaohai
 * @ClassName SysLogConsumer
 * @description: 系统操作日志消费者
 * @date 2025年03月19日
 * @version: 1.0.0
 */
@Service
@Slf4j
public class SysLogConsumer {

    private final ISysLogService sysLogService;
    private final ObjectMapper objectMapper;
    private final LogExtRepository logExtRepository;

    public SysLogConsumer(ISysLogService sysLogService, LogExtRepository logExtRepository) {
        this.sysLogService = sysLogService;
        this.objectMapper = new ObjectMapper();
        // 注册 JavaTimeModule 支持 LocalDate 和 LocalDateTime 序列化
        this.objectMapper.registerModule(new JavaTimeModule());
        this.logExtRepository = logExtRepository;
    }

    @KafkaListener(topics = "${syslog.kafka.topic}", groupId = "syslog-group")
    public void consumeSysLog(ConsumerRecord<String, String> record) {
        try {
            // 获取消息内容
            String message = record.value();
            SysLogDto sysLogDto = objectMapper.readValue(message, SysLogDto.class);

            // 先将 请求参数、响应结果、异常信息保存到mongodb里
            LogExt logExt = new LogExt();
            logExt.setRequestParams(sysLogDto.getRequestParams());
            logExt.setResponseResult(sysLogDto.getResponseResult());
            logExt.setExceptionInfo(sysLogDto.getExceptionInfo());
            LogExt savedLogExt = logExtRepository.save(logExt);

            // 将SysLogDto转换为SysLog对象
            SysLog sysLog = SysLog.builder()
                    .userName(sysLogDto.getUserName())
                    .callDate(sysLogDto.getCallDate())
                    .callTime(sysLogDto.getCallTime())
                    .url(sysLogDto.getUrl())
                    .executionTime(sysLogDto.getExecutionTime())
                    .result(sysLogDto.getResult())
                    .createdTime(sysLogDto.getCreatedTime()).build();

            // 将mongodb里保存号的数据的id赋值给sysLog里的extId字段
            sysLog.setExtId(savedLogExt.getId());
            sysLogService.save(sysLog);
            log.info("SysLog persisted successfully: {}", sysLog);
        } catch (Exception e) {
            log.error("Failed to consume or persist SysLog, message content: {}", record.value());
        }
    }

}
